package app.function;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import app.bean.CategoryInsightResult;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * 品类洞察处理函数 - 实现实时洞察分析
 */
public class CategoryInsightProcessFunction extends KeyedProcessFunction<String, CategoryInsightResult, CategoryInsightResult> {
    
    private ListState<CategoryInsightResult> categoryListState;
    private static final String DATA_SOURCE = "kafka_dwd_market_product";
    private static final Integer VERSION = 1;
    
    public CategoryInsightProcessFunction() {
        // 移除了未使用的analysisType参数
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        categoryListState = getRuntimeContext().getListState(
            new ListStateDescriptor<>("categoryListState", 
                TypeInformation.of(new TypeHint<CategoryInsightResult>() {}))
        );
    }
    
    @Override
    public void processElement(CategoryInsightResult value, Context ctx, Collector<CategoryInsightResult> out) throws Exception {
        // 添加到状态列表
        categoryListState.add(value);
        
        // 注册定时器，延迟1秒处理分析
        ctx.timerService().registerEventTimeTimer(ctx.timerService().currentWatermark() + 1000);
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<CategoryInsightResult> out) throws Exception {
        // 获取所有品类数据
        List<CategoryInsightResult> categoryList = new ArrayList<>();
        for (CategoryInsightResult category : categoryListState.get()) {
            categoryList.add(category);
        }
        
        // 计算总销售额用于市场份额计算
        double totalMarketSales = categoryList.stream()
                .mapToDouble(c -> c.getTotalSalesAmount() != null ? c.getTotalSalesAmount() : 0.0)
                .sum();
        
        // 为每个品类计算洞察指标
        for (CategoryInsightResult category : categoryList) {
            // 计算市场份额
            category.calculateMarketShare(totalMarketSales);
            
            // 计算平均价格
            category.calculateAvgPrice();
            
            // 设置竞争等级
            category.setCompetitionLevel();
            
            // 设置实时数仓标准字段
            category.setCurrentTimestamp();
            category.setDataLineage(DATA_SOURCE, VERSION);
            category.setPeriod("7d"); // 7天统计周期
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long windowStart = timestamp - 604800000; // 7天前
            category.setWindowTime(sdf.format(new Date(windowStart)), sdf.format(new Date(timestamp)));
            
            out.collect(category);
        }
        
        // 清空状态
        categoryListState.clear();
    }
}